package org.netwia.networks;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.builder.ToStringBuilder;
import org.netwia.utils.DomainUtils;
import org.netwia.utils.StartStopService;
import org.netwia.utils.ValidationUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Simple asynchronous network processor.
 * It creates new thread and searches for the node where there are all inputs available.
 * If this node is found, then processing is invoked and output is pushed to the listeners.
 * Then processing of the next node continues.
 * 
 * @author radek.hecl
 *
 * @param <T> type of the network data
 */
public class AsyncSimpleNetworkProcessor<T> implements StartStopService {

    /**
     * Logger object.
     */
    private static final Logger logger = LoggerFactory.getLogger(AsyncSimpleNetworkProcessor.class);
    
    /**
     * Builder object.
     *
     * @param <T> type of the network data
     */
    public static class Builder<T> {

        /**
         * Network which will be processed.
         */
        private Network<T> network;

        /**
         * Child services which are started right after processing and stopped right before processing.
         */
        private Set<StartStopService> childServices = new HashSet<StartStopService>();

        /**
         * Sets network which will be processed.
         * 
         * @param network network
         * @return this instance
         */
        public Builder<T> setNetwork(Network<T> network) {
            this.network = network;
            return this;
        }

        /**
         * Sets child services. These are started right after processing and stopped right before processing.
         * 
         * @param childServices child services
         * @return this instance
         */
        public Builder<T> setChildServices(Collection<StartStopService> childServices) {
            this.childServices = DomainUtils.softCopySet(childServices);
            return this;
        }
        
        /**
         * Adds child service.
         * 
         * @param service child service
         * @return this instance
         */
        public Builder<T> addChildService(StartStopService service) {
            childServices.add(service);
            return this;
        }
        
        /**
         * Builds the result object.
         * 
         * @return created object
         */
        public AsyncSimpleNetworkProcessor<T> build() {
            return new AsyncSimpleNetworkProcessor<T>(this);
        }
    }

    /**
     * Network which will be processed.
     */
    private Network<T> network;
    /**
     * Child services which are started right after processing and stopped right before processing.
     */
    private Set<StartStopService> childServices;

    /**
     * Lock object.
     */
    private Lock lock = new ReentrantLock();

    /**
     * Flag indicating whether processing is running or not.
     */
    private boolean running = false;

    /**
     * Thread in which the processing is running.
     */
    private Thread processingThread = null;

    /**
     * Creates new instance.
     * 
     * @param builder builder object
     */
    public AsyncSimpleNetworkProcessor(Builder<T> builder) {
        network = builder.network;
        childServices = Collections.unmodifiableSet(DomainUtils.softCopySet(builder.childServices));
        guardInvariants();
    }

    /**
     * Guards this object to be consistent. Throws exception if this is not the case.
     */
    private void guardInvariants() {
        ValidationUtils.guardNotNull(network, "network cannot be null");
        ValidationUtils.guardNotNullCollection(childServices, "childServices cannot have null element");
    }

    @Override
    public void start() {
        lock.lock();
        try {
            if (running) {
                return;
            }
            processingThread = new Thread(new ProcessingRunnable<T>(this));
            processingThread.start();
            running = true;
            for (StartStopService service : childServices) {
                service.start();
            }
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void stop() {
        lock.lock();
        try {
            if (!running) {
                return;
            }
            for (StartStopService service : childServices) {
                service.stop();
            }
            running = false;
            processingThread.join();
            processingThread = null;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public String toString() {
        return ToStringBuilder.reflectionToString(this);
    }

    /**
     * Inner class invokable in the separate thread.
     * 
     * @param <T> type of the network data 
     */
    private static class ProcessingRunnable<T> implements Runnable {

        /**
         * Processor.
         */
        private AsyncSimpleNetworkProcessor<T> processor;

        /**
         * Creates new instance.
         * 
         * @param processor processor
         */
        public ProcessingRunnable(AsyncSimpleNetworkProcessor<T> processor) {
            this.processor = processor;
            guardInvariants();
        }

        /**
         * Guards this object to be consistent. Throws exception if this is not the case.
         */
        private void guardInvariants() {
            ValidationUtils.guardNotNull(processor, "processor cannot be null");
        }

        @Override
        public void run() {
            Network<T> network = processor.network;
            Set<String> nodes = network.getNodeNames();
            while (processor.running) {
                for (String node : nodes) {
                    Map<String, InputProvider<T>> inputProviders = network.getInputs(node);
                    boolean ok = true;
                    for (String inputName : inputProviders.keySet()) {
                        if (!inputProviders.get(inputName).isNextInputReady()) {
                            ok = false;
                            break;
                        }
                    }
                    if (!ok) {
                        continue;
                    }
                    Map<String, T> inputs = new HashMap<String, T>();
                    for (String inputName : inputProviders.keySet()) {
                        inputs.put(inputName, inputProviders.get(inputName).getNextInput());
                    }
                    T output = null;
                    try {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Starting node: name = " + node + "; inputs = " + inputs);
                        }
                        output = network.getNode(node).process(inputs);
                        if (logger.isDebugEnabled()) {
                            logger.debug("Finished node: name = " + node + "; output = " + output);
                        }
                    } catch (Exception e) {
                        throw new RuntimeException("error in node: node = " + node, e);
                    }
                    for (OutputListener<T> listener : network.getOutputs(node)) {
                        listener.outputProduced(output);
                    }
                }
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        
        @Override
        public String toString() {
            return ToStringBuilder.reflectionToString(this);
        }
    }
}
